package com.kongming.kmdm.ws.rest.endpoint;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.routing.SmallestMailboxPool;
import com.alibaba.fastjson.JSONObject;
import com.kongming.kmdm.common.mq.MQSubscribePoint;
import com.kongming.kmdm.ws.list.WSList;
import com.kongming.kmdm.ws.rest.actor.PushMsgActor;
import com.kongming.kmdm.ws.rest.auth.UserProperty;
import com.kongming.kmdm.ws.rest.coder.SubscribeMsgDecoder;
import com.kongming.kmdm.ws.rest.coder.SubscribeMsgEncoder;
import com.kongming.kmdm.ws.rest.config.WSConfig;
import com.kongming.kmdm.ws.rest.msg.SubRequestMsg;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.*;

/**
 * websocket的subscribe服务，主要提供监听服务器返回消息
 *
 * @author Alex
 * @des
 * @date 2015年4月30日
 */
@ServerEndpoint(value = "/subscribe", decoders = {
        SubscribeMsgDecoder.class
}, encoders = {
        SubscribeMsgEncoder.class
}, configurator = WebSocketServerConfig.class)
public class WebSocketServerEndpoint {

    private static ActorRef actor;

    @SuppressWarnings("unused")
    private static MQSubscribePoint mqSub;

    private static WSConfig config;

    private static Map<String, WSList> ssnMap;

    private static Map<String, String> idMap;

    private static Object mapObject = new Object();

    //private static List <Session>               sessionList = null;
    private static Object obj = new Object();

    private static boolean isInit = false;

    private static ScheduledExecutorService service = Executors.newScheduledThreadPool(2);

    private static Logger logger = LoggerFactory.getLogger(WebSocketServerEndpoint.class);

    //private IRedisCache                         cache;

    /**
     * 新session加入
     *
     * @param session
     */
    @OnOpen
    public void onOpen(Session session, EndpointConfig endConfig) {

        //初始化
        //生成actor处理器
        synchronized (obj) {
            if (!isInit) {
                config = new WSConfig();
                //cache = new RedisStorage (config.getRedisURl ());
                ssnMap = new ConcurrentHashMap<String, WSList>();
                idMap = new ConcurrentHashMap<String, String>();
                ActorSystem system = ActorSystem.create("WebSocketServer");
                actor = system.actorOf(Props.create(PushMsgActor.class, ssnMap).withRouter(
                        new SmallestMailboxPool(config.getActorNum())));
                mqSub = new MQSubscribePoint(config.getExchangeName(), config.getRoutingKey(),
                        config.getQueueName(), actor);

                service.scheduleWithFixedDelay(new Runnable() {
                    @Override
                    public void run() {
                        mqSub.check();
                    }
                }, 0, 1, TimeUnit.MINUTES);
                isInit = true;
            }
        }

        // addSession ("test", session);
        String sessionID = session.getRequestParameterMap().get("sid").get(0);
        //userID,前台传过来
        //String userID = session.getRequestParameterMap ().get ("kmUserId").get (0);
        UserProperty user = (UserProperty) endConfig.getUserProperties().get(sessionID);

        if (user.getIsValid()) {
            //注册mq监听事件
            logger.info("success linked {} id-{} kmUserId {}", sessionID, session.getId(), user.getKmUserId());
            addSession(user.getKmUserId(), session);
            //addSession (sessionID, session);
        } else {
            JSONObject json = new JSONObject();
            json.put("status", "success");
            json.put("type", "Discard");
            json.put("msg", new JSONObject());
            logger.info("send Discard msg");
            try {
                session.getBasicRemote().sendObject(json.toJSONString());
                session.close();
                logger.info("closed***********");
            } catch (IOException | EncodeException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            //            logger.info ("closed***********");
            //            try {
            //                session.close ();
            //            }
            //            catch (IOException e) {
            //                logger.error (e);
            //                e.printStackTrace ();
            //            }
        }
    }

    /**
     * 收取前端消息
     *
     * @param msg
     * @param session
     * @return
     */
    @OnMessage
    public void onMessage(SubRequestMsg msg, Session session) {

        switch (msg.getCmd()) {
            //        case QUIT:
            //            try {
            //                session.close (new CloseReason (CloseCodes.NORMAL_CLOSURE, "websocket closed"));
            //                delSession (session);
            //            }
            //            catch (IOException e) {
            //                throw new RuntimeException (e);
            //            }
            //            break;
            case Ping:
                try {
                    JSONObject json = new JSONObject();
                    json.put("status", "success");
                    json.put("type", "Pong");
                    json.put("msg", new JSONObject());
                    logger.info("send Pong msg");
                    session.getBasicRemote().sendObject(json.toJSONString());
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (EncodeException e) {
                    e.printStackTrace();
                }
                break;

            //        case RECIVEMSG:
            //            try {
            //                //删除REDIS中的消息
            //                String userId = session.getRequestParameterMap ().get ("kmUserId").get (0);
            //                cache.delete (config.getRedisMsgKey () + userId);
            //            }
            //            catch (Exception e) {
            //                e.printStackTrace ();
            //            }
            //            break;
            default:
                break;
        }
    }

    @OnClose
    public void onClose(Session session, CloseReason closeReason) {

        delSession(session);
    }

    /**
     * 添加对应的key到监听
     *
     * @param key
     * @param session
     */
    private void addSession(String key, Session session) {

        WSList sessionList = ssnMap.get(key);
        if (sessionList == null) {
            sessionList = new WSList();
            ssnMap.put(key, sessionList);
        }
        logger.info("addSession {} to key {}; ssnMapHashCode {}", session.getId(), key, ssnMap.hashCode());
//        for (String mapKey : ssnMap.keySet()) {
//            logger.info("ssnMap key {} size {}", mapKey, ssnMap.get(mapKey).getSsnSize());
//        }
        sessionList.add(session);
        idMap.put(session.getId(), key);
        //output new listInfo
        //        for (Session index : sessionList) {
        //            Log.info ("now ssnList include:" + index.getId ());
        //        }
    }

    /**
     * 删除对应key的session
     */
    private void delSession(Session session) {

        WSList wsList = null;
        synchronized (mapObject) {
            String key = idMap.get(session.getId());
            if (key != null) {
                wsList = ssnMap.get(key);
            }
        }
        if (wsList != null) {
            wsList.remove(session);
        }
        idMap.remove(session.getId());
    }
}
